package com.jiayuan.cn.energy.flink.mqtt;

import com.jiayuan.cn.energy.util.MqttClientUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.StringData;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

@Slf4j
public class MqttSourceFunction extends RichSourceFunction {

    private final String[] words = new String[]{
            "Welcome", "to", "the", "world", "of", "flink"
    };
    @Override
    public void run(SourceContext ctx) throws Exception {
//        Random random = new Random();
//        while (true) {
//            int index = random.nextInt(words.length);
//            GenericRowData row = new GenericRowData(1);
//            row.setField(0,StringData.fromString(words[index]));
//            ctx.collect(row);
//            Thread.sleep(100);
//        }

        String  broker = "wss://a5450752.ala.cn-hangzhou.emqxsl.cn:8084";
        String clientId = "JavaSample";
        String  username = "m1";
        String password = "1111";
        String topic = "/v1/report/+/model";
        String content = "Hello MQTTm 666";
        int qos = 2;

        MqttClientUtil mqttClientUtil = new MqttClientUtil(broker, clientId, username, password);
        mqttClientUtil.connect();
//        mqttClientUtil.publish(topic, content, qos);
         BlockingQueue<String> queue = new ArrayBlockingQueue<String>(22);
        mqttClientUtil.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                System.out.println("Connection lost");
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
//                    System.out.println("Message arrived: " + new String(message.getPayload()));
//                log.info("Message arrived:{} ", new String(message.getPayload()));
                queue.put(new String(message.getPayload()));
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                System.out.println("Delivery complete");
            }
        });

        mqttClientUtil.subscribe(topic, qos);
        Thread.sleep(5000); // Wait for messages
        log.info("startttttt");
        //使用阻塞队列的好处是队列空的时候程序会一直阻塞到这里不会浪费CPU资源
//        log.info(queue.take());

        while (true) {
            String str  = queue.take();
//            log.info("received: {}", str);
            GenericRowData row = new GenericRowData(1);
            row.setField(0, StringData.fromString(str));
            ctx.collect(row);
        }

//        while (true) {
//            int index = random.nextInt(words.length);
//            String str  = queue.take();
//            GenericRowData row = new GenericRowData(1);
//            row.setField(0,StringData.fromString(str));
//            ctx.collect(row);
//            Thread.sleep(100);
//        }
    }
    @Override
    public void cancel() {
    }
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }


}